package cn.jly.flink.table_sql;

import cn.jly.flink.entity.SensorReading;
import cn.jly.flink.utils.FlinkUtils;
import com.alibaba.fastjson.JSON;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;

import static org.apache.flink.table.api.Expressions.$;

/**
 * flink1.11之后底层默认使用blink的planner，1.10之前使用的老版本的planner
 *
 * @PackageName cn.jly.flink.table_sql
 * @ClassName TableAndSqlDemo01
 * @Description table api和sql入门示例
 * @Author 姬岚洋
 * @Date 2021/2/5 下午 4:34
 */
public class TableAndSqlDemo01_port {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = FlinkUtils.getStreamExecutionEnv();
        env.setParallelism(1);

        final ParameterTool parameterTool = ParameterTool.fromArgs(args);
        final String host = parameterTool.get("host", "node01");
        final int port = parameterTool.getInt("port", 9999);

        // 读取数据
        // final DataStreamSource<String> inputStream = env.readTextFile("d:/SensorReading.txt");
        final DataStreamSource<String> inputStream = env.socketTextStream(host, port);

        // 转换成javaBean
        final SingleOutputStreamOperator<SensorReading> sensorReadingDataStream = inputStream.flatMap(new FlatMapFunction<String, SensorReading>() {
            @Override
            public void flatMap(String value, Collector<SensorReading> out) throws Exception {
                if (StringUtils.isNotEmpty(value)) {
                    out.collect(JSON.parseObject(value, SensorReading.class));
                }
            }
        });

        // 创建表环境
        final StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // table api -> 基于数据流创建表
        final Table table = tableEnv.fromDataStream(sensorReadingDataStream);
        // 调用table api进行转换操作
        final Table resultTable = table.select($("id"), $("temperature"))
                .where($("id").isEqual("1001"));

        // flink sql -> 执行sql
        tableEnv.createTemporaryView("tbl_sensor", sensorReadingDataStream);
        String sql = "select id, temperature from tbl_sensor where id = '1001'";
        final Table resultSqlTable = tableEnv.sqlQuery(sql);

        // 输出
        // 打印结构
        resultTable.printSchema();
        // 转换成流
        final DataStream<Row> resultStream = tableEnv.toAppendStream(resultTable, Row.class);
        resultStream.print("tableApi");

        resultSqlTable.printSchema();
        final DataStream<Row> resultSqlStream = tableEnv.toAppendStream(resultSqlTable, Row.class);
        resultSqlStream.print("sql");

        FlinkUtils.executeStream(env, "FirstDemo");
    }
}
